package gupao.concurrency.threads;

import lombok.RequiredArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class RunnableTaskDemo {

    static final boolean daemonFalg = false;
    public static void main(String[] args) {
        BizTaskProcessor processor = new BizTaskProcessor();
        processor.start();
    }

    @Setter
    @RequiredArgsConstructor
    static class MyTask implements Runnable {
        final BizTaskGenerator taskGenerator;
        volatile boolean stop = false;
        @Override
        public void run() {
            //你的业务代码
            Thread t = Thread.currentThread();
            /*t.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
                @Override
                public void uncaughtException(Thread t, Throwable e) {
                    //统一处理未捕捉的异常
                }
            });*/

            while(!stop){
                Integer taskIndex = taskGenerator.getNextTaskIndex();
                try{
                    if(taskIndex != null){
                        doBizLogic(taskIndex);
                    }
                } catch (IllegalTaskException e){
                    //继续执行
                    log.warn("ignore IllegalTaskException, {} continue", t.getName());
                } catch ( TaskRanOutException e ){
                    //执行结束
                    log.info("task ran out, stop biz processor {}", t.getName());
                    stop = true;
                }
                catch ( InterruptedException e ){
                    log.error("{} interrupted, stop it",t.getName());
                    Thread.currentThread().interrupt();
                }
                catch ( Throwable e ){
                    recordExceptionTask(e, taskIndex);
                }
            }
            beforeComplete();
        }

        void beforeComplete(){
            log.info("{} before complete..", Thread.currentThread().getName());
            if( taskGenerator != null ){
                taskGenerator.interrupt();
            }
        }

        void doBizLogic(int taskId) throws InterruptedException {
            //模拟异常: 任务不合法
            if( taskId % 15 == 0){
                throw new IllegalTaskException();
            }

            //模拟异常: 任务执行结束
            if(taskId == 32){
                throw new TaskRanOutException();
            }

            log.info("{} process task {}", Thread.currentThread().getName(), taskId);
        }

        void recordExceptionTask(Throwable t, int taskIndex){
            log.error("just log exception task {}, exception: {}", taskIndex, t.getMessage());
        }
    }

    @RequiredArgsConstructor
    static class BizTaskProcessor {
        final BizTaskGenerator taskGenerator;
        final MyTask bizTask;
        volatile Thread threadRef;

        public BizTaskProcessor() {
            taskGenerator = new BizTaskGenerator();
            bizTask = new MyTask(taskGenerator);
        }

        public void start() {
            taskGenerator.start();

            threadRef = new Thread(bizTask, "biz-task-processor");
            threadRef.setDaemon(daemonFalg);
            threadRef.start();
        }

        public void stop(){
            bizTask.setStop(true);
            threadRef.interrupt();
        }
    }

    static class BizTaskGenerator {
        ArrayBlockingQueue<Integer> taskQueue = new ArrayBlockingQueue<>(10000);
        AtomicInteger taskIndex = new AtomicInteger(0);
        volatile Thread t;

        public void interrupt(){
            if( t != null ){
                t.interrupt();
            }
        }

        public Integer getNextTaskIndex(){
            try {
                return taskQueue.take();
            } catch (Exception e) {
                //ignore
                return null;
            }
        }

        //模拟没3秒钟产生一个task
        public void start() {
            t = new Thread(() -> {
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        taskQueue.put(taskIndex.incrementAndGet());
                        log.info("{} generate a new task {}", Thread.currentThread().getName(), taskIndex.get());
                        Thread.sleep(300);
                    } catch (InterruptedException e) {
                        log.info("{} interrupted, will stop soon", Thread.currentThread().getName());
                        Thread.currentThread().interrupt();
                    }
                }
            }, "biz-task-generator");
            t.start();
        }
    }

    static class IllegalTaskException extends RuntimeException{
    }

    static class TaskRanOutException extends RuntimeException{
    }

    static class TaskUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler{
        @Override
        public void uncaughtException(Thread t, Throwable e) {

        }
    }
}
